package com.linkstec.quartz.source;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.HashMap;

import org.bson.Document;
import org.quartz.DisallowConcurrentExecution;
import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.linkstec.mongo.MongoDbUtil;
import com.linkstec.util.ConfigUtil;
import com.mongodb.client.FindIterable;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoCursor;
import com.mongodb.client.MongoDatabase;
import com.mongodb.client.model.Filters;

@DisallowConcurrentExecution
public class PushJob implements Job {
	private static Logger logger = LoggerFactory.getLogger(PushJob.class);
	private static MongoDatabase mongoDatabase = MongoDbUtil.getMongoDatabase(
			ConfigUtil.getProperty("mongo.database", "MOT"));
	private static MongoCollection<Document> pushJobConfigCollection = MongoDbUtil.getMongoCollection(
			mongoDatabase, "PUSH_JOB_CONFIG");
	@Override
	public void execute(JobExecutionContext context) throws JobExecutionException {
		logger.info("开始本次推送！！！");

		String jobName = context.getJobDetail().getDescription();
		logger.info(context.getJobDetail().getJobDataMap() + "");

		InetAddress addr = null;
		try {
			addr = InetAddress.getLocalHost();
		} catch (UnknownHostException e1) {
			e1.printStackTrace();
		}
        String hostName=addr.getHostName().toString(); //获取本机计算机名称
		MongoCursor<Document> mongoCursor = null;
        FindIterable<Document> findIterable = pushJobConfigCollection.find(Filters.eq("serverid", hostName));
        mongoCursor = findIterable.iterator();
        while(mongoCursor.hasNext()){
        	Document jobConfig = mongoCursor.next();
        	if(jobName.equals(jobConfig.getString("jobName"))){
        		String jobParams = jobConfig.getString("jobParams");

        		HashMap<String, String> dtoInput = new HashMap<String, String>();
        		for (String jobParam : jobParams.split(" ")){
    				String key = jobParam.split("=")[0];
    				String value = jobParam.split("=")[1];
    				dtoInput.put(key, value);
    			}
//        		GtjaCftEventFlowPushOneKafkaService pushService = new GtjaCftEventFlowPushOneKafkaService();
//        		pushService.execute(dtoInput);
        	}
        }
	}
}